package com.practice.gmall.realtime.sink;

import com.alibaba.fastjson.JSONObject;
import com.practice.gmall.realtime.bean.TableProcess;
import com.practice.gmall.realtime.common.Constant;
import com.practice.gmall.realtime.util.JdbcUtil;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;

public class PhoenixSink extends RichSinkFunction<Tuple2<TableProcess, JSONObject>> {

    private Connection conn;
    private PreparedStatement stmt;

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        conn = JdbcUtil.getJdbcConn(Constant.PHOENIX_JDBC_DRIVER, Constant.PHOENIX_DB_URL, null, null);
    }

    @Override
    public void close() throws Exception {
        super.close();

        if (stmt != null) {
            stmt.close();
        }

        if (conn != null) {
            conn.close();
        }
    }

    @Override
    public void invoke(Tuple2<TableProcess, JSONObject> value, Context context) throws Exception {
        super.invoke(value, context);

        TableProcess tp = value.f0;
        JSONObject data = value.f1;
        String operateType = data.getString("operate_type");

        if ("delete".equals(operateType)) {
            deleteLine(data, tp);
        } else {
            insertLine(data, tp);
        }

    }

    private void deleteLine(JSONObject data, TableProcess tp) throws SQLException {
        StringBuilder sql = new StringBuilder();
        sql.append("DELETE FROM ")
                .append(tp.getSinkTable())
                .append(" WHERE id = ?");

        System.out.println(sql);

        stmt = conn.prepareStatement(sql.toString());
        stmt.setString(1,tp.getSinkPk());
        stmt.execute();
        conn.commit();
    }

    private void insertLine(JSONObject data, TableProcess tp) throws SQLException {
        StringBuilder sql = new StringBuilder();

        String skCols = tp.getSinkColumns();
        sql.append("UPSERT INTO ")
                .append(tp.getSinkTable())
                .append("(").append(skCols).append(") ")
                .append("VALUES").append("(").append(skCols.replaceAll("[^,]+", "?")).append(")");

        //System.out.println("PhoenixSink.invoke " + sql.toString() );
        stmt = conn.prepareStatement(sql.toString());

        String[] cols = skCols.split(",");

        for (int i = 0; i < cols.length; i++) {
            stmt.setString(i + 1, data.getString(cols[i]));
        }

        stmt.execute();
        conn.commit();
    }

}

